package com.intct.dws;

import com.intct.common.FlinkSqlWithUtil;
import com.intct.func.RowKeyFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-23 14:10
 * HBase创建：namespace -> dws    如：create_namespace 'dws'
 *           table --> dws_vehicle_distr  如：create 'dws:dws_vehicle_distr', 'info'
 */
public class DwsVecicleDistrSQL {
    public static void main(String[] args) {
        // 1 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置全局并行度
        env.setParallelism(1);
        // 开启检查点，并且设置检查为一次语义
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);
        // TableAPI环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 2 source映射 -- kafka
        // 样例：
        // {
        //  "action_time": 1758074384,
        //  "area_idarea_id": "110106",
        //  "car_shelf_uumber": "Y696799102D",
        //  "dese_car_shelf_number": "Y69*****02D",
        //  "car": "56",
        //  "area_name": "北京市丰台区"
        //}
        tenv.executeSql("CREATE TABLE flink_dws_vehicle (                    " +
                "  action_time TIMESTAMP(3)                                           " +
                "  ,area_idarea_id STRING                                             " +
                "  ,car STRING                                                        " +
                "  ,area_name STRING                                                  " +
                "  ,pt AS PROCTIME()                                                  " +
                "  ,WATERMARK FOR action_time AS action_time - INTERVAL '5' SECOND    " +
                ") " + FlinkSqlWithUtil.getKafkaSourceWith("dwd_vehicle", "dwd_vehicle_group", "dwd_vehicle_client_id"));

        // 3 映射DWS(HBase)
        tenv.executeSql("CREATE TABLE hbase_dws_vehicle_distr (   " +
                " id STRING                                                " +
                " ,info ROW<                                               " +
                "    window_start STRING                                   " +
                "    , window_end STRING                                   " +
                "    , area_idarea_id STRING                               " +
                "    , area_name STRING                                    " +
                "    , num_vechicle STRING                                 " +
                "    >                                                     " +
                " )                                                        " +
                FlinkSqlWithUtil.getHBaseSinkWith("dws", "dws_vehicle_distr"));

        // 4 保存结果
        // 4.1 注册函数--时间戳(yyyy-MM-dd HH:mm:ss.SSS)转换时间戳(BIGINT)
        tenv.createTemporarySystemFunction("C_TO_TS", RowKeyFunction.class);

        // 4.1 聚合后，插入HBase
        tenv.executeSql("INSERT INTO hbase_dws_vehicle_distr      " +
                " SELECT                                                   " +
                "   CONCAT(C_TO_TS(window_start), '*', area_idarea_id)     " +
                "   ,ROW(                                                  " +
                "      CAST(window_start AS STRING)                        " +
                "      ,CAST(window_end AS STRING)                         " +
                "      ,area_idarea_id                                     " +
                "      ,area_name                                          " +
                "      ,CAST(COUNT(DISTINCT car) AS STRING)                " +
                "    ) AS info                                             " +
                " FROM TABLE(TUMBLE(TABLE flink_dws_vehicle, DESCRIPTOR(action_time), INTERVAL '1' MINUTES))" +
                " GROUP BY                                                 " +
                "    window_start                                          " +
                "    , window_end                                          " +
                "    , area_idarea_id                                      " +
                "    , area_name                                           ");
    }
}
